package storm;

import java.io.File;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;
import org.apache.storm.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Tomcat 日志统计
 * 
 * @author 李岩飞
 * @email eliyanfei@126.com
 * @date 2018年6月7日下午2:54:35
 */
public class StormTopology {
	private static Logger LOG = LoggerFactory.getLogger(StormTopology.class);

	public static void main(String[] args) throws Exception {
		TopologyBuilder builder = new TopologyBuilder();
		builder.setSpout("log", new LogSpout());
		builder.setBolt("ip", new LogIPBolt()).shuffleGrouping("log");
		builder.setBolt("url", new LogURLBolt()).shuffleGrouping("ip");
		builder.setBolt("word", new WordBolt(), 5).fieldsGrouping("url", new Fields("word"));

		Config conf = new Config();
		conf.setDebug(false);

		if (args != null && args.length > 0) {
			conf.setNumWorkers(3);
			LOG.info(args[0] + "===============");
			LOG.info(args[1] + "=================");
			conf.put("path", args[1]);
			StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createTopology());
		} else {
			LocalCluster cluster = new LocalCluster();
			cluster.submitTopology("test3", conf, builder.createTopology());
			Utils.sleep(20000);
			cluster.killTopology("test3");
			cluster.shutdown();
		}
	}
}